package DianShang_2024.ds_03.clean

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, current_timestamp, date_format, lit, to_timestamp}

import java.text.SimpleDateFormat
import java.util.Date

object clean06 {
  def main(args: Array[String]): Unit = {
    /*
        6、将ods库中order_detail表昨天的分区（子任务一中生成的分区）数据抽取到dwd库中fact_order_detail的动态分区表，分区字段为etl_date，
        类型为String，取create_time值并将格式转换为yyyyMMdd，并添加dwd_insert_user、dwd_insert_time、dwd_modify_user、
        dwd_modify_time四列，其中dwd_insert_user、dwd_modify_user均填写“user1”，dwd_insert_time、dwd_modify_time均填写当前操作
        时间，并进行数据类型转换。使用hive cli执行show partitions dwd.fact_order_detail命令，将结果截图粘贴至客户端桌面【Release\任务B
        提交结果.docx】中对应的任务序号下。
     */

    //  这里和第五题一样，需要先去hive的客户端修改create_time字段的数据类型为string,否则写入不进去，因为to_timestamp和to_date函数都自带格式
    //  因为yyyyMMdd并不是规范的时间类型格式



    //  准备环境
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("数据清洗第六题")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .enableHiveSupport()
      .getOrCreate()

    spark.sql("use dwd03")

    spark.sql("drop table if exists dwd03.fact_order_")
    spark.sql(
      """
        |create table if not exists dwd03.fact_order_detail(
        |id int,
        |order_id int,
        |sku_id int,
        |sku_name string,
        |img_url string,
        |order_price decimal(10,2),
        |sku_num string,
        |create_time timestamp,
        |source_type string,
        |source_id int,
        |dwd_insert_user string,
        |dwd_insert_time timestamp,
        |dwd_modify_user string,
        |dwd_modify_time timestamp
        |)
        |partitioned  by(etl_date string)
        |""".stripMargin)


    //  1.使用sql的方式

    //  读取ods层的数据，并进行处理
//    spark.sql(
//      """
//        |select
//        |id,
//        |order_id,
//        |sku_id,
//        |sku_name,
//        |img_url,
//        |order_price,
//        |sku_num,
//        |date_format(create_time,"yyyyMMdd") as create_time,
//        |source_type,
//        |source_id,
//        |"user1" as dwd_insert_user,
//        |to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss") as dwd_insert_time,
//        |"user1" as dwd_modify_user,
//        |to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss") as dwd_modify_time,
//        |etl_date
//        |from ods03.order_detail
//        |""".stripMargin).createOrReplaceTempView("result")
//
//    //  插入数据
//    spark.sql(
//      """
//        |insert overwrite table dwd03.fact_order_detail
//        |partition(etl_date)
//        |select * from result
//        |""".stripMargin)


    //  2.使用dataframe的形式
    spark.sql("select * from ods03.order_detail")
      .where(col("etl_date")==="20240101")
      .withColumn("create_time",date_format(col("create_time"),"yyyyMMdd"))
      .withColumn("dwd_insert_user", lit("user1"))
      .withColumn(
        "dwd_insert_time",
        lit(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())).cast("timestamp")
      )
      .withColumn("dwd_modify_user", lit("user1"))
      .withColumn(
        "dwd_modify_time",
        lit(to_timestamp(date_format(current_timestamp(), "yyyy-MM-dd HH:mm:ss"), "yyyy-MM-dd HH:mm:ss"))
      )
      .write.mode("append")
      .format("hive")
      .partitionBy("etl_date")
      .saveAsTable("fact_order_detail")


    //  关闭环境
    spark.close()
  }

}
